/**   
 * @Title: AvroUtil.java 
 * @Package com.org.source.avro 
 * @Description: TODO(用一句话描述该文件做什么) 
 * @author cssuger@163.com   
 * @date 2016年12月21日 上午10:19:58 
 * @version V1.0   
 */
package com.bicon.base.data.common.exchange;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Collection;
import java.util.List;

import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericData.Record;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.Encoder;
import org.apache.avro.io.EncoderFactory;

import com.bicon.base.data.common.StringTools;
import com.bicon.base.data.common.UUIDGenerator;

import avro.shaded.com.google.common.collect.Lists;



/**
 * @ClassName: AvroUtil
 * @Description: 可否将数据交换的利用avro进行统一的数据交换
 * @author cssuger@163.com
 * @date 2016年12月21日 上午10:19:58
 */
public class AvroUtil {

	private ResourceLoad resourceLioad;
	
	

	
	public AvroUtil() {

	}
	
	public AvroUtil(ResourceLoad resourceLioad) {
        this.resourceLioad = resourceLioad;
	}

	

	/**
	 * 
	 * @Title: parseProtocol
	 * @Description: 解析协议
	 * @param @param file
	 * @param @return 设定文件
	 * @return Protocol 返回类型
	 * @throws
	 */
	public Protocol parseProtocol() {
		return resourceLioad.loadProtocol();
	}

	
	
	
	

	/**
	 * 
	 * @Title: getAllSchemas
	 * @Description: 获取所有的type数据结构
	 * @param @return 设定文件
	 * @return List<Schema> 返回类型
	 * @throws
	 */
	public List<Schema> getAllSchemas() {
		Protocol protocol = this.parseProtocol();
		if (null == protocol) {
			throw new NullPointerException("protocol is nou null");
		}
		Collection<Schema> schemas = protocol.getTypes();
		return Lists.newArrayList(schemas);
	}
	
	/**
	 * 
	 * @Title: put   王结构里面增加数据
	 * @Description: TODO(这里用一句话描述这个方法的作用)   
	 * @param: @param protocol
	 * @param: @param name
	 * @param: @param values      
	 * @return: void      
	 * @throws
	 */
	public String put(Schema schema,String [] values) {
		GenericRecord requestData = new GenericData.Record(schema);
		for(int i =0;i<values.length;i++) {
			requestData.put(i, StringTools.mkString(values[i]));
		}
		return requestData.toString();
	}
	
	 

	/**
	 * 
	 * @Title: getSchemasById
	 * @Description: "types": [ {"name":"LogEntity", "type":"record", "fields":[
	 *               {"name":"body", "type":["string", "null"]},
	 *               {"name":"timestamp", "type":["string", "null"]},
	 *               {"name":"header", "type":["string", "null"]}
	 * 
	 *               ]},
	 * 
	 *               ],
	 * @param typeName 表示一个record的唯一标示,上面的LogEntity就是typeName
	 * @param @return 设定文件
	 * @return Schema 返回类型
	 * @throws
	 */
	public Schema getSchemasById(String typeName) {
		Protocol protocol = this.parseProtocol();
		if (null == protocol) {
			throw new NullPointerException("protocol is nou null");
		}
		Schema entitySchema = protocol.getType(typeName);
		return entitySchema;
	}
	
	public Collection<Schema> findAllSchemas(){
		Protocol protocol = this.parseProtocol();
		if (null == protocol) {
			throw new NullPointerException("protocol is nou null");
		}
		return protocol.getTypes();
	}

	/**
	 * 
	 * @Title: getAllFieldBySchema
	 * @Description: 获取某个
	 * @param @param typeName
	 * @param @return 设定文件
	 * @return List<Field> 返回类型
	 * @throws
	 */
	public List<Field> getAllFieldBySchema(String typeName) {
		Schema entitySchema = this.getSchemasById(typeName);
		return entitySchema.getFields();
	}

	/**
	 * 
	 * @Title: serializing 
	 * @Description: 这种的是序列化到文件里面 
	 * @param @param schema
	 * @param @param genericrecord
	 * @param @param file
	 * @param @throws IOException    设定文件 
	 * @return void    返回类型 
	 * @throws
	 */
	public void serializing(Schema schema, Record genericrecord, File file)
			throws IOException {

		DatumWriter<GenericRecord> userDatumWriter = new GenericDatumWriter<GenericRecord>();
		OutputStream outputStream = new ByteArrayOutputStream();
		Encoder encoder = EncoderFactory.get().jsonEncoder(schema, outputStream);
		userDatumWriter.write(genericrecord, encoder);
		DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<GenericRecord>(
				userDatumWriter);
		DataFileWriter<GenericRecord> dfw = null;
		if (!file.exists()) {
			dfw = dataFileWriter.create(schema, file);
		} else {
			dfw = dataFileWriter.appendTo(file);
		}
		try {
			dfw.append(genericrecord);

		} catch (Exception e) {
			e.printStackTrace();
			// logger.error("write local file [" + filename + "] error", e);
		} finally {
			dfw.close();
			dataFileWriter.close();
		}

	}

	
	
	public void deserializing(Schema schema, File file) throws IOException {
		DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(
				schema);
		DataFileReader<GenericRecord> dataFileReader = new DataFileReader<GenericRecord>(
				file, datumReader);
		GenericRecord user = null;
		try{
			while (dataFileReader.hasNext()) {
				// Reuse user object by passing it to next(). This saves us from
				// allocating and garbage collecting many objects for files with
				// many items.
				user = dataFileReader.next(user);
				System.out.println(user);
				
			}
		}finally{
			dataFileReader.close();
		}
		
	}

	public void writeToLocalFile(Schema schema, Record genericrecord)
			throws IOException {

	}

	public static void main(String[] args) throws IOException {
		String str = "{\r\n" + 
				"    \"namespace\": \"com.bicon.datamanager.log.avro\",\r\n" + 
				"    \"protocol\": \"BusinessLogProtocol\",\r\n" + 
				"    \"doc\": \"用户业务日志数据结构\",\r\n" + 
				"    \"name\": \"BusinessLog\",\r\n" + 
				" \r\n" + 
				"    \"types\": [\r\n" + 
				"        {\"name\":\"BusinessLog\", \"type\":\"record\",\r\n" + 
				"          \"fields\":[\r\n" + 
				"                {\"name\":\"id\",              \"type\":[\"string\", \"null\"]},\r\n" + 
				"                {\"name\":\"operater\",        \"type\":[\"string\", \"null\"]},\r\n" + 
				"                {\"name\":\"bus_name\",        \"type\":[\"string\", \"null\"]},\r\n" + 
				"                {\"name\":\"befor_hander\",    \"type\":[\"string\", \"null\"]},\r\n" + 
				"                {\"name\":\"key\",             \"type\":[\"string\", \"null\"]},\r\n" + 
				"                {\"name\":\"result\",          \"type\":[\"string\", \"null\"]},\r\n" + 
				"				{\"name\":\"reson\" ,          \"type\":[\"string\", \"null\"]},\r\n" + 
				"				{\"name\":\"datetime\" ,       \"type\":[\"string\", \"null\"]}\r\n" + 
				"				\r\n" + 
				"             ]\r\n" + 
				"		},\r\n" + 
				"		{\"name\":\"HearBeat\", \"type\":\"record\",\r\n" + 
				"          \"fields\":[\r\n" + 
				"                {\"name\":\"ping\",  \"type\":[\"string\", \"null\"]},\r\n" + 
				"				{\"name\":\"responseMsg\" ,\"type\":[\"string\", \"null\"]}\r\n" + 
				"            ]\r\n" + 
				"		}\r\n" + 
				"       \r\n" + 
				"        \r\n" + 
				"    ],\r\n" + 
				" \r\n" + 
				"    \"messages\":    {\r\n" + 
				"        \"sendLogger\":{\r\n" + 
				"            \"doc\" : \"发送日志方法\",\r\n" + 
				"			\"name\": \"logger\",\r\n" + 
				"            \"request\" :[{\"name\":\"businessLog\",\"type\":\"BusinessLog\" }],\r\n" + 
				"            \"response\" :\"BusinessLog\"\r\n" + 
				"        } , \r\n" + 
				"        \"sendhearBeat\":{\r\n" + 
				"            \"doc\" : \"心跳\",\r\n" + 
				"			\"name\": \"hear\",\r\n" + 
				"            \"request\" :[{\"name\":\"hearbeat\",\"type\":\"HearBeat\" }],\r\n" + 
				"            \"response\" :\"HearBeat\"\r\n" + 
				"        }		\r\n" + 
				"    }   \r\n" + 
				"}";
		AvroUtil util = new AvroUtil(new StringResourceLoad(str));
		
		String [] values = {new UUIDGenerator().generate().toString(),"ADMIN","规则编辑","1111111111111111","123454","1","1",""};
		System.out.println(util.put(util.getSchemasById("BusinessLog"), values));
		

	}
}
